package com.distributer.report.infrastructure.strom.bolt;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import com.alibaba.fastjson.JSONObject;
import com.distributer.report.domain.statistics.model.ProductOrderStatistics;
import com.distributer.report.domain.statistics.valueobj.OrderValueObj;

/**
 * 根据产品维度统计每个产品产生的订单数量、订单金额
 * 每个消息都有一个唯一的消息id，消息id通过kafkaData生成，已经处理消息id存在redis中，
 * 先判断消息id是否已经被处理过，处理过的消息id不再重复处理（每个消息id需要有过期时间，过期时间根据消息的最大重试时间而定）。
 * 消息处理失败日志需要存储到数据库中，便于后期错误跟踪。
 * @author xiaolin
 *
 */
public class ProductStatisticsBolt extends BaseRichBolt{

	private static final long serialVersionUID = -6910331074483905260L;
	private static final Logger LOGGER = LoggerFactory.getLogger(ProductStatisticsBolt.class);
	
	private OutputCollector collector;
	private static ConcurrentHashMap<String, ProductOrderStatistics> orderStatistics = new ConcurrentHashMap<String, ProductOrderStatistics>();;
	private JedisPool jedisPool;
	private static final String ORDER_STATISTIC_PRODUCT = "ORDER_STATISTIC_PRODUCT";
	
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		initRedis();
	}
	
	@Override
	public void execute(Tuple input) {
		OrderValueObj orderValueObj = (OrderValueObj)input.getValueByField("orderValueObj");
		try {
			int i = 1/0;
			//不包含说明没有处理过，下一步需要处理
			if (!include(orderValueObj.getMessageId())) {
				ProductOrderStatistics productOrderStatistics = orderStatistics.get(orderValueObj.getProductCode());
				if (productOrderStatistics == null) {
					productOrderStatistics = new ProductOrderStatistics(orderValueObj.getProductCode(),orderValueObj.getProductName());
					orderStatistics.put(orderValueObj.getProductCode(), productOrderStatistics);
				}
				productOrderStatistics.statistics(orderValueObj);
				LOGGER.info(Thread.currentThread().getName()+" ProductStatisticsBolt 产品编码 : "+ orderValueObj.getProductCode()+" 订单的统计结果："+JSONObject.toJSONString(orderStatistics.get(orderValueObj.getProductCode())));
				//在redis中添加一条数据并设置数据过期时间 
				setMessagIdToRedis(orderValueObj);
			}
			//发送到下一个bolt
			collector.emit(input, new Values(orderValueObj));
			collector.ack(input);
			LOGGER.info(Thread.currentThread().getName()+" ProductStatisticsBolt ack => messageID:"+orderValueObj.getMessageId());
		} catch (Exception e) {
			collector.fail(input);
			LOGGER.info(Thread.currentThread().getName()+" ProductStatisticsBolt fail => messageID:"+orderValueObj.getMessageId(),e);
		}
	}
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("orderValueObj"));
	}
	
	private void setMessagIdToRedis(OrderValueObj orderValueObj){
		Jedis jedis = jedisPool.getResource();
		try{
			String key = ORDER_STATISTIC_PRODUCT+":"+orderValueObj.getMessageId();
			jedis.set(key, orderValueObj.getMessageId());
			jedis.expire(key, 60);
		}finally{
			jedis.close();
		}
	}

	private boolean include(String messageId){
		Jedis jedis = jedisPool.getResource();
		try{
			String key = ORDER_STATISTIC_PRODUCT+":"+messageId;
			return jedis.exists(key);
		}finally{
			jedis.close();
		}
	}
	
	private void initRedis(){
		JedisPoolConfig config = new JedisPoolConfig();
		config.setMaxTotal(100);
		config.setMaxIdle(5);
		config.setMaxWaitMillis(1000 * 10); 
		config.setTestOnBorrow(true);
		this.jedisPool = new JedisPool(config, "192.168.57.134", 6379);
	}
}
